package cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.receiveMQ;

import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.data.RocketMQArgvData;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.log4j.Logger;

import java.text.ParseException;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * Created by Black-Laptop on 7/11/2017.
 */
public class ReceiveMQ {
    public static Logger LOG = Logger.getLogger(ReceiveMQ.class);

    private String topic = "";
    private String group = "";
    private String nameserver = "";

    private REReceiveSessionBuilder builder = null;

    private ArrayBlockingQueue<REMessage> queue;

    public ReceiveMQ(RocketMQArgvData rocketMQArgvData, ArrayBlockingQueue<REMessage> queue) throws REConnectionException, ParseException {
        this.topic = rocketMQArgvData.getReceiveConsumerTopic();
        this.group = rocketMQArgvData.getReceiveConsumerGroup();
        this.nameserver = rocketMQArgvData.getReceiveConsumerNameserver();
        this.queue = queue;
        init();
    }

    public REReceiveSession start() {
        REReceiveSession session = null;
        try {
            session = (REReceiveSession) builder.build();
            session.start();
        } catch (REConnectionException e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
        }

        return session;
    }

    private void init() {
        REConnection conn = new REConnection(nameserver);
        LOG.info("Init nameserver: " + nameserver);

        try {
            builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(topic);
            LOG.info("Init topic: " + topic);
        } catch (REConnectionException e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
        }
        LOG.info("Init Position: last");
        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
        builder.setHandler(new MessageSuccessReceiver(queue));
        builder.setFailureHandler(new MessageFailReceiver());
        builder.setGroupName(group);
        LOG.info("Init group: " + group);
        builder.setConsumeThreadNum(1);

    }
}
